package com.charles.bigdata.mapreduce.order;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class OrderMain {

	public static void main(String[] args) throws Exception {
		// 1.创建一个job和任务入口
		Job job = Job.getInstance(new Configuration());
		job.setJarByClass(OrderMain.class); // main方法所在的class

		// 2.指定job的mapper和输出的类型<k2 v2>
		job.setMapperClass(OrderMapper.class);// 指定Mapper类
		job.setMapOutputKeyClass(OrderBean.class); // k2的类型
		job.setMapOutputValueClass(NullWritable.class); // v2的类型

		// 3.指定job的reducer和输出的类型<k4  v4>
		job.setReducerClass(OrderReducer.class);// 指定Reducer类
		job.setOutputKeyClass(OrderBean.class); // k4的类型
		job.setOutputValueClass(NullWritable.class); // v4的类型

		job.setGroupingComparatorClass(MyGroupingComparator.class);
		job.setPartitionerClass(ItemIdPartitioner.class);
		job.setNumReduceTasks(2);

		// 4.指定job的输入和输出
		FileInputFormat.setInputPaths(job, new Path("E:\\temp\\orders.txt")); // 待统计的数据问题
		FileOutputFormat.setOutputPath(job, new Path("E:\\temp\\orders")); // 输出路径

		// 5.执行job
		job.waitForCompletion(true);
	}
}